package kn.topicMsgTransfer.mq;


import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONException;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.rabbitmq.client.Channel;
import kn.topicMsgTransfer.common.Constants;
import kn.topicMsgTransfer.entity.TopicOrgan;
import kn.topicMsgTransfer.req.EventPlanData;
import kn.topicMsgTransfer.req.TopicPushQueueTempCacheDTO;
import kn.topicMsgTransfer.service.TopicOrganService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.Objects;

@Component
@Slf4j
public class Consumer {



    @Autowired
    private TopicOrganService topicOrganService;

    @Autowired
    private RedisTemplate<String,Object> redisTemplate;


    @Autowired
    @Qualifier("topicMqRabbitTemplate")
    private RabbitTemplate topicMqRabbitTemplate;

    @RabbitListener(queues = Constants.topic_push_7d_msg_que0,containerFactory = "sevenDayMqFactory")
    public void keywordReceiveMessage0(String message, Channel channel, Message rabbitMessage) {

        try {
            EventPlanData eventPlanData = JSON.parseObject(message, EventPlanData.class);
            // 根据push7dId查询话题id
            LambdaQueryWrapper<TopicOrgan> topicOrganLambdaQueryWrapper = Wrappers.<TopicOrgan>lambdaQuery().eq(TopicOrgan::getPush7dId, eventPlanData.getTopicID());
            TopicOrgan topicOrgan = topicOrganService.getOne(topicOrganLambdaQueryWrapper);
            if(Objects.isNull(topicOrgan)){
                channel.basicAck(rabbitMessage.getMessageProperties().getDeliveryTag(), false);
                    return;
            }
            eventPlanData.setIs7d(1);
            Integer id = topicOrgan.getId();
            String queueName = Constants.TOPIC_PUSH_QUEUE + id;
            // 查询缓存中的队列名是否存在 并且可投递 是则投递临时队列
            String obj = String.valueOf(redisTemplate.opsForHash().get(Constants.TOPIC_PUSH_QUEUE_TEMP, queueName));
            TopicPushQueueTempCacheDTO topicPushQueueTempCacheDTO = null;
            if (StrUtil.isNotBlank(obj)) {
                ObjectMapper objectMapper = new ObjectMapper();
                try {
                    topicPushQueueTempCacheDTO = objectMapper.readValue(obj, TopicPushQueueTempCacheDTO.class);
                } catch (Exception e) {
                    // 处理异常
                    e.printStackTrace();
                    log.error("【临时队列数据序列化失败】：{}",e.getMessage());
                }
                if (Objects.nonNull(topicPushQueueTempCacheDTO) && topicPushQueueTempCacheDTO.getIsDelivery()) {
                    String queueName1 = topicPushQueueTempCacheDTO.getQueueName();
                    topicMqRabbitTemplate.convertAndSend(queueName1,JSONUtil.toJsonStr(eventPlanData));
                    log.info("【转发处理】id：{}投递临时队列{}",id,queueName1);

                    channel.basicAck(rabbitMessage.getMessageProperties().getDeliveryTag(), false);
                    return;
                }
            }
            // 否则投递正常队列
            String routingKey = Constants.topic_push_7d_msg_que_transfer + id % 10;
            topicMqRabbitTemplate.convertAndSend(routingKey,JSONUtil.toJsonStr(eventPlanData));
            log.info("【转发处理】id：{}投递正常队列{}",id,routingKey);
            channel.basicAck(rabbitMessage.getMessageProperties().getDeliveryTag(), false);
        }catch (JSONException jsonException){
            log.error("json解析异常，消息: {}" ,JSONUtil.toJsonStr(message));
        }catch (Exception e) {
            e.printStackTrace();
            try {
                int retryCount = (int) rabbitMessage.getMessageProperties().getHeaders().getOrDefault("x-death", 0);
                if (retryCount > 5) {
                    // 重试次数超过阈值，选择丢弃
                    channel.basicNack(rabbitMessage.getMessageProperties().getDeliveryTag(), false, false);
                    log.error("消息推送数据重试次数超过阈值，丢弃消息: {}" + rabbitMessage.getMessageProperties().getMessageId());
                } else {
                    // 否则继续重试
                    channel.basicNack(rabbitMessage.getMessageProperties().getDeliveryTag(), false, true);
                }
            } catch (IOException ex) {
                ex.printStackTrace();
            }
        }
    }

    @RabbitListener(queues = Constants.topic_push_7d_msg_que1,containerFactory = "sevenDayMqFactory")
    public void keywordReceiveMessage1(String message, Channel channel, Message rabbitMessage) {

        try {
            EventPlanData eventPlanData = JSON.parseObject(message, EventPlanData.class);
            // 根据push7dId查询话题id
            LambdaQueryWrapper<TopicOrgan> topicOrganLambdaQueryWrapper = Wrappers.<TopicOrgan>lambdaQuery().eq(TopicOrgan::getPush7dId, eventPlanData.getTopicID());
            TopicOrgan topicOrgan = topicOrganService.getOne(topicOrganLambdaQueryWrapper);
            if(Objects.isNull(topicOrgan)){
                channel.basicAck(rabbitMessage.getMessageProperties().getDeliveryTag(), false);
                    return;
            }
            eventPlanData.setIs7d(1);
            Integer id = topicOrgan.getId();
            String queueName = Constants.TOPIC_PUSH_QUEUE + id;
            // 查询缓存中的队列名是否存在 并且可投递 是则投递临时队列
            String obj = String.valueOf(redisTemplate.opsForHash().get(Constants.TOPIC_PUSH_QUEUE_TEMP, queueName));
            TopicPushQueueTempCacheDTO topicPushQueueTempCacheDTO = null;
            if (StrUtil.isNotBlank(obj)) {
                ObjectMapper objectMapper = new ObjectMapper();
                try {
                    topicPushQueueTempCacheDTO = objectMapper.readValue(obj, TopicPushQueueTempCacheDTO.class);
                } catch (Exception e) {
                    // 处理异常
                    e.printStackTrace();
                    log.error("【临时队列数据序列化失败】：{}",e.getMessage());
                }
                if (Objects.nonNull(topicPushQueueTempCacheDTO) && topicPushQueueTempCacheDTO.getIsDelivery()) {
                    String queueName1 = topicPushQueueTempCacheDTO.getQueueName();
                    topicMqRabbitTemplate.convertAndSend(queueName1,JSONUtil.toJsonStr(eventPlanData));
                    log.info("【转发处理】id：{}投递临时队列{}",id,queueName1);

                    channel.basicAck(rabbitMessage.getMessageProperties().getDeliveryTag(), false);
                    return;
                }
            }
            // 否则投递正常队列
            String routingKey = Constants.topic_push_7d_msg_que_transfer + id % 10;
            topicMqRabbitTemplate.convertAndSend(routingKey,JSONUtil.toJsonStr(eventPlanData));
            log.info("【转发处理】id：{}投递正常队列{}",id,routingKey);
            channel.basicAck(rabbitMessage.getMessageProperties().getDeliveryTag(), false);
        }catch (JSONException jsonException){
            log.error("json解析异常，消息: {}" ,JSONUtil.toJsonStr(message));
        }catch (Exception e) {
            e.printStackTrace();
            try {
                int retryCount = (int) rabbitMessage.getMessageProperties().getHeaders().getOrDefault("x-death", 0);
                if (retryCount > 5) {
                    // 重试次数超过阈值，选择丢弃
                    channel.basicNack(rabbitMessage.getMessageProperties().getDeliveryTag(), false, false);
                    log.error("消息推送数据重试次数超过阈值，丢弃消息: {}" + rabbitMessage.getMessageProperties().getMessageId());
                } else {
                    // 否则继续重试
                    channel.basicNack(rabbitMessage.getMessageProperties().getDeliveryTag(), false, true);
                }
            } catch (IOException ex) {
                ex.printStackTrace();
            }
        }
    }

    @RabbitListener(queues = Constants.topic_push_7d_msg_que2,containerFactory = "sevenDayMqFactory")
    public void keywordReceiveMessage2(String message, Channel channel, Message rabbitMessage) {

        try {
            EventPlanData eventPlanData = JSON.parseObject(message, EventPlanData.class);
            // 根据push7dId查询话题id
            LambdaQueryWrapper<TopicOrgan> topicOrganLambdaQueryWrapper = Wrappers.<TopicOrgan>lambdaQuery().eq(TopicOrgan::getPush7dId, eventPlanData.getTopicID());
            TopicOrgan topicOrgan = topicOrganService.getOne(topicOrganLambdaQueryWrapper);
            if(Objects.isNull(topicOrgan)){
                channel.basicAck(rabbitMessage.getMessageProperties().getDeliveryTag(), false);
                    return;
            }
            eventPlanData.setIs7d(1);
            Integer id = topicOrgan.getId();
            String queueName = Constants.TOPIC_PUSH_QUEUE + id;
            // 查询缓存中的队列名是否存在 并且可投递 是则投递临时队列
            String obj = String.valueOf(redisTemplate.opsForHash().get(Constants.TOPIC_PUSH_QUEUE_TEMP, queueName));
            TopicPushQueueTempCacheDTO topicPushQueueTempCacheDTO = null;
            if (StrUtil.isNotBlank(obj)) {
                ObjectMapper objectMapper = new ObjectMapper();
                try {
                    topicPushQueueTempCacheDTO = objectMapper.readValue(obj, TopicPushQueueTempCacheDTO.class);
                } catch (Exception e) {
                    // 处理异常
                    e.printStackTrace();
                    log.error("【临时队列数据序列化失败】：{}",e.getMessage());
                }
                if (Objects.nonNull(topicPushQueueTempCacheDTO) && topicPushQueueTempCacheDTO.getIsDelivery()) {
                    String queueName1 = topicPushQueueTempCacheDTO.getQueueName();
                    topicMqRabbitTemplate.convertAndSend(queueName1,JSONUtil.toJsonStr(eventPlanData));
                    log.info("【转发处理】id：{}投递临时队列{}",id,queueName1);

                    channel.basicAck(rabbitMessage.getMessageProperties().getDeliveryTag(), false);
                    return;
                }
            }
            // 否则投递正常队列
            String routingKey = Constants.topic_push_7d_msg_que_transfer + id % 10;
            topicMqRabbitTemplate.convertAndSend(routingKey,JSONUtil.toJsonStr(eventPlanData));
            log.info("【转发处理】id：{}投递正常队列{}",id,routingKey);
            channel.basicAck(rabbitMessage.getMessageProperties().getDeliveryTag(), false);
        }catch (JSONException jsonException){
            log.error("json解析异常，消息: {}" ,JSONUtil.toJsonStr(message));
        }catch (Exception e) {
            e.printStackTrace();
            try {
                int retryCount = (int) rabbitMessage.getMessageProperties().getHeaders().getOrDefault("x-death", 0);
                if (retryCount > 5) {
                    // 重试次数超过阈值，选择丢弃
                    channel.basicNack(rabbitMessage.getMessageProperties().getDeliveryTag(), false, false);
                    log.error("消息推送数据重试次数超过阈值，丢弃消息: {}" + rabbitMessage.getMessageProperties().getMessageId());
                } else {
                    // 否则继续重试
                    channel.basicNack(rabbitMessage.getMessageProperties().getDeliveryTag(), false, true);
                }
            } catch (IOException ex) {
                ex.printStackTrace();
            }
        }
    }

    @RabbitListener(queues = Constants.topic_push_7d_msg_que3,containerFactory = "sevenDayMqFactory")
    public void keywordReceiveMessage3(String message, Channel channel, Message rabbitMessage) {

        try {
            EventPlanData eventPlanData = JSON.parseObject(message, EventPlanData.class);
            // 根据push7dId查询话题id
            LambdaQueryWrapper<TopicOrgan> topicOrganLambdaQueryWrapper = Wrappers.<TopicOrgan>lambdaQuery().eq(TopicOrgan::getPush7dId, eventPlanData.getTopicID());
            TopicOrgan topicOrgan = topicOrganService.getOne(topicOrganLambdaQueryWrapper);
            if(Objects.isNull(topicOrgan)){
                channel.basicAck(rabbitMessage.getMessageProperties().getDeliveryTag(), false);
                    return;
            }
            eventPlanData.setIs7d(1);
            Integer id = topicOrgan.getId();
            String queueName = Constants.TOPIC_PUSH_QUEUE + id;
            // 查询缓存中的队列名是否存在 并且可投递 是则投递临时队列
            String obj = String.valueOf(redisTemplate.opsForHash().get(Constants.TOPIC_PUSH_QUEUE_TEMP, queueName));
            TopicPushQueueTempCacheDTO topicPushQueueTempCacheDTO = null;
            if (StrUtil.isNotBlank(obj)) {
                ObjectMapper objectMapper = new ObjectMapper();
                try {
                    topicPushQueueTempCacheDTO = objectMapper.readValue(obj, TopicPushQueueTempCacheDTO.class);
                } catch (Exception e) {
                    // 处理异常
                    e.printStackTrace();
                    log.error("【临时队列数据序列化失败】：{}",e.getMessage());
                }
                if (Objects.nonNull(topicPushQueueTempCacheDTO) && topicPushQueueTempCacheDTO.getIsDelivery()) {
                    String queueName1 = topicPushQueueTempCacheDTO.getQueueName();
                    topicMqRabbitTemplate.convertAndSend(queueName1,JSONUtil.toJsonStr(eventPlanData));
                    log.info("【转发处理】id：{}投递临时队列{}",id,queueName1);

                    channel.basicAck(rabbitMessage.getMessageProperties().getDeliveryTag(), false);
                    return;
                }
            }
            // 否则投递正常队列
            String routingKey = Constants.topic_push_7d_msg_que_transfer + id % 10;
            topicMqRabbitTemplate.convertAndSend(routingKey,JSONUtil.toJsonStr(eventPlanData));
            log.info("【转发处理】id：{}投递正常队列{}",id,routingKey);
            channel.basicAck(rabbitMessage.getMessageProperties().getDeliveryTag(), false);
        }catch (JSONException jsonException){
            log.error("json解析异常，消息: {}" ,JSONUtil.toJsonStr(message));
        }catch (Exception e) {
            e.printStackTrace();
            try {
                int retryCount = (int) rabbitMessage.getMessageProperties().getHeaders().getOrDefault("x-death", 0);
                if (retryCount > 5) {
                    // 重试次数超过阈值，选择丢弃
                    channel.basicNack(rabbitMessage.getMessageProperties().getDeliveryTag(), false, false);
                    log.error("消息推送数据重试次数超过阈值，丢弃消息: {}" + rabbitMessage.getMessageProperties().getMessageId());
                } else {
                    // 否则继续重试
                    channel.basicNack(rabbitMessage.getMessageProperties().getDeliveryTag(), false, true);
                }
            } catch (IOException ex) {
                ex.printStackTrace();
            }
        }
    }


    @RabbitListener(queues = Constants.topic_push_7d_msg_que4,containerFactory = "sevenDayMqFactory")
    public void keywordReceiveMessage4(String message, Channel channel, Message rabbitMessage) {

        try {
            EventPlanData eventPlanData = JSON.parseObject(message, EventPlanData.class);
            // 根据push7dId查询话题id
            LambdaQueryWrapper<TopicOrgan> topicOrganLambdaQueryWrapper = Wrappers.<TopicOrgan>lambdaQuery().eq(TopicOrgan::getPush7dId, eventPlanData.getTopicID());
            TopicOrgan topicOrgan = topicOrganService.getOne(topicOrganLambdaQueryWrapper);
            if(Objects.isNull(topicOrgan)){
                channel.basicAck(rabbitMessage.getMessageProperties().getDeliveryTag(), false);
                    return;
            }
            eventPlanData.setIs7d(1);
            Integer id = topicOrgan.getId();
            String queueName = Constants.TOPIC_PUSH_QUEUE + id;
            // 查询缓存中的队列名是否存在 并且可投递 是则投递临时队列
            String obj = String.valueOf(redisTemplate.opsForHash().get(Constants.TOPIC_PUSH_QUEUE_TEMP, queueName));
            TopicPushQueueTempCacheDTO topicPushQueueTempCacheDTO = null;
            if (StrUtil.isNotBlank(obj)) {
                ObjectMapper objectMapper = new ObjectMapper();
                try {
                    topicPushQueueTempCacheDTO = objectMapper.readValue(obj, TopicPushQueueTempCacheDTO.class);
                } catch (Exception e) {
                    // 处理异常
                    e.printStackTrace();
                    log.error("【临时队列数据序列化失败】：{}",e.getMessage());
                }
                if (Objects.nonNull(topicPushQueueTempCacheDTO) && topicPushQueueTempCacheDTO.getIsDelivery()) {
                    String queueName1 = topicPushQueueTempCacheDTO.getQueueName();
                    topicMqRabbitTemplate.convertAndSend(queueName1,JSONUtil.toJsonStr(eventPlanData));
                    log.info("【转发处理】id：{}投递临时队列{}",id,queueName1);

                    channel.basicAck(rabbitMessage.getMessageProperties().getDeliveryTag(), false);
                    return;
                }
            }
            // 否则投递正常队列
            String routingKey = Constants.topic_push_7d_msg_que_transfer + id % 10;
            topicMqRabbitTemplate.convertAndSend(routingKey,JSONUtil.toJsonStr(eventPlanData));
            log.info("【转发处理】id：{}投递正常队列{}",id,routingKey);
            channel.basicAck(rabbitMessage.getMessageProperties().getDeliveryTag(), false);
        }catch (JSONException jsonException){
            log.error("json解析异常，消息: {}" ,JSONUtil.toJsonStr(message));
        }catch (Exception e) {
            e.printStackTrace();
            try {
                int retryCount = (int) rabbitMessage.getMessageProperties().getHeaders().getOrDefault("x-death", 0);
                if (retryCount > 5) {
                    // 重试次数超过阈值，选择丢弃
                    channel.basicNack(rabbitMessage.getMessageProperties().getDeliveryTag(), false, false);
                    log.error("消息推送数据重试次数超过阈值，丢弃消息: {}" + rabbitMessage.getMessageProperties().getMessageId());
                } else {
                    // 否则继续重试
                    channel.basicNack(rabbitMessage.getMessageProperties().getDeliveryTag(), false, true);
                }
            } catch (IOException ex) {
                ex.printStackTrace();
            }
        }
    }


    @RabbitListener(queues = Constants.topic_push_7d_msg_que5,containerFactory = "sevenDayMqFactory")
    public void keywordReceiveMessage5(String message, Channel channel, Message rabbitMessage) {

        try {
            EventPlanData eventPlanData = JSON.parseObject(message, EventPlanData.class);
            // 根据push7dId查询话题id
            LambdaQueryWrapper<TopicOrgan> topicOrganLambdaQueryWrapper = Wrappers.<TopicOrgan>lambdaQuery().eq(TopicOrgan::getPush7dId, eventPlanData.getTopicID());
            TopicOrgan topicOrgan = topicOrganService.getOne(topicOrganLambdaQueryWrapper);
            if(Objects.isNull(topicOrgan)){
                channel.basicAck(rabbitMessage.getMessageProperties().getDeliveryTag(), false);
                    return;
            }
            eventPlanData.setIs7d(1);
            Integer id = topicOrgan.getId();
            String queueName = Constants.TOPIC_PUSH_QUEUE + id;
            // 查询缓存中的队列名是否存在 并且可投递 是则投递临时队列
            String obj = String.valueOf(redisTemplate.opsForHash().get(Constants.TOPIC_PUSH_QUEUE_TEMP, queueName));
            TopicPushQueueTempCacheDTO topicPushQueueTempCacheDTO = null;
            if (StrUtil.isNotBlank(obj)) {
                ObjectMapper objectMapper = new ObjectMapper();
                try {
                    topicPushQueueTempCacheDTO = objectMapper.readValue(obj, TopicPushQueueTempCacheDTO.class);
                } catch (Exception e) {
                    // 处理异常
                    e.printStackTrace();
                    log.error("【临时队列数据序列化失败】：{}",e.getMessage());
                }
                if (Objects.nonNull(topicPushQueueTempCacheDTO) && topicPushQueueTempCacheDTO.getIsDelivery()) {
                    String queueName1 = topicPushQueueTempCacheDTO.getQueueName();
                    topicMqRabbitTemplate.convertAndSend(queueName1,JSONUtil.toJsonStr(eventPlanData));
                    log.info("【转发处理】id：{}投递临时队列{}",id,queueName1);

                    channel.basicAck(rabbitMessage.getMessageProperties().getDeliveryTag(), false);
                    return;
                }
            }
            // 否则投递正常队列
            String routingKey = Constants.topic_push_7d_msg_que_transfer + id % 10;
            topicMqRabbitTemplate.convertAndSend(routingKey,JSONUtil.toJsonStr(eventPlanData));
            log.info("【转发处理】id：{}投递正常队列{}",id,routingKey);
            channel.basicAck(rabbitMessage.getMessageProperties().getDeliveryTag(), false);
        }catch (JSONException jsonException){
            log.error("json解析异常，消息: {}" ,JSONUtil.toJsonStr(message));
        }catch (Exception e) {
            e.printStackTrace();
            try {
                int retryCount = (int) rabbitMessage.getMessageProperties().getHeaders().getOrDefault("x-death", 0);
                if (retryCount > 5) {
                    // 重试次数超过阈值，选择丢弃
                    channel.basicNack(rabbitMessage.getMessageProperties().getDeliveryTag(), false, false);
                    log.error("消息推送数据重试次数超过阈值，丢弃消息: {}" + rabbitMessage.getMessageProperties().getMessageId());
                } else {
                    // 否则继续重试
                    channel.basicNack(rabbitMessage.getMessageProperties().getDeliveryTag(), false, true);
                }
            } catch (IOException ex) {
                ex.printStackTrace();
            }
        }
    }

    @RabbitListener(queues = Constants.topic_push_7d_msg_que6,containerFactory = "sevenDayMqFactory")
    public void keywordReceiveMessage6(String message, Channel channel, Message rabbitMessage) {

        try {
            EventPlanData eventPlanData = JSON.parseObject(message, EventPlanData.class);
            // 根据push7dId查询话题id
            LambdaQueryWrapper<TopicOrgan> topicOrganLambdaQueryWrapper = Wrappers.<TopicOrgan>lambdaQuery().eq(TopicOrgan::getPush7dId, eventPlanData.getTopicID());
            TopicOrgan topicOrgan = topicOrganService.getOne(topicOrganLambdaQueryWrapper);
            if(Objects.isNull(topicOrgan)){
                channel.basicAck(rabbitMessage.getMessageProperties().getDeliveryTag(), false);
                    return;
            }
            eventPlanData.setIs7d(1);
            Integer id = topicOrgan.getId();
            String queueName = Constants.TOPIC_PUSH_QUEUE + id;
            // 查询缓存中的队列名是否存在 并且可投递 是则投递临时队列
            String obj = String.valueOf(redisTemplate.opsForHash().get(Constants.TOPIC_PUSH_QUEUE_TEMP, queueName));
            TopicPushQueueTempCacheDTO topicPushQueueTempCacheDTO = null;
            if (StrUtil.isNotBlank(obj)) {
                ObjectMapper objectMapper = new ObjectMapper();
                try {
                    topicPushQueueTempCacheDTO = objectMapper.readValue(obj, TopicPushQueueTempCacheDTO.class);
                } catch (Exception e) {
                    // 处理异常
                    e.printStackTrace();
                    log.error("【临时队列数据序列化失败】：{}",e.getMessage());
                }
                if (Objects.nonNull(topicPushQueueTempCacheDTO) && topicPushQueueTempCacheDTO.getIsDelivery()) {
                    String queueName1 = topicPushQueueTempCacheDTO.getQueueName();
                    topicMqRabbitTemplate.convertAndSend(queueName1,JSONUtil.toJsonStr(eventPlanData));
                    log.info("【转发处理】id：{}投递临时队列{}",id,queueName1);

                    channel.basicAck(rabbitMessage.getMessageProperties().getDeliveryTag(), false);
                    return;
                }
            }
            // 否则投递正常队列
            String routingKey = Constants.topic_push_7d_msg_que_transfer + id % 10;
            topicMqRabbitTemplate.convertAndSend(routingKey,JSONUtil.toJsonStr(eventPlanData));
            log.info("【转发处理】id：{}投递正常队列{}",id,routingKey);
            channel.basicAck(rabbitMessage.getMessageProperties().getDeliveryTag(), false);
        }catch (JSONException jsonException){
            log.error("json解析异常，消息: {}" ,JSONUtil.toJsonStr(message));
        }catch (Exception e) {
            e.printStackTrace();
            try {
                int retryCount = (int) rabbitMessage.getMessageProperties().getHeaders().getOrDefault("x-death", 0);
                if (retryCount > 5) {
                    // 重试次数超过阈值，选择丢弃
                    channel.basicNack(rabbitMessage.getMessageProperties().getDeliveryTag(), false, false);
                    log.error("消息推送数据重试次数超过阈值，丢弃消息: {}" + rabbitMessage.getMessageProperties().getMessageId());
                } else {
                    // 否则继续重试
                    channel.basicNack(rabbitMessage.getMessageProperties().getDeliveryTag(), false, true);
                }
            } catch (IOException ex) {
                ex.printStackTrace();
            }
        }
    }

    @RabbitListener(queues = Constants.topic_push_7d_msg_que7,containerFactory = "sevenDayMqFactory")
    public void keywordReceiveMessage7(String message, Channel channel, Message rabbitMessage) {

        try {
            EventPlanData eventPlanData = JSON.parseObject(message, EventPlanData.class);
            // 根据push7dId查询话题id
            LambdaQueryWrapper<TopicOrgan> topicOrganLambdaQueryWrapper = Wrappers.<TopicOrgan>lambdaQuery().eq(TopicOrgan::getPush7dId, eventPlanData.getTopicID());
            TopicOrgan topicOrgan = topicOrganService.getOne(topicOrganLambdaQueryWrapper);
            if(Objects.isNull(topicOrgan)){
                channel.basicAck(rabbitMessage.getMessageProperties().getDeliveryTag(), false);
                    return;
            }
            eventPlanData.setIs7d(1);
            Integer id = topicOrgan.getId();
            String queueName = Constants.TOPIC_PUSH_QUEUE + id;
            // 查询缓存中的队列名是否存在 并且可投递 是则投递临时队列
            String obj = String.valueOf(redisTemplate.opsForHash().get(Constants.TOPIC_PUSH_QUEUE_TEMP, queueName));
            TopicPushQueueTempCacheDTO topicPushQueueTempCacheDTO = null;
            if (StrUtil.isNotBlank(obj)) {
                ObjectMapper objectMapper = new ObjectMapper();
                try {
                    topicPushQueueTempCacheDTO = objectMapper.readValue(obj, TopicPushQueueTempCacheDTO.class);
                } catch (Exception e) {
                    // 处理异常
                    e.printStackTrace();
                    log.error("【临时队列数据序列化失败】：{}",e.getMessage());
                }
                if (Objects.nonNull(topicPushQueueTempCacheDTO) && topicPushQueueTempCacheDTO.getIsDelivery()) {
                    String queueName1 = topicPushQueueTempCacheDTO.getQueueName();
                    topicMqRabbitTemplate.convertAndSend(queueName1,JSONUtil.toJsonStr(eventPlanData));
                    log.info("【转发处理】id：{}投递临时队列{}",id,queueName1);

                    channel.basicAck(rabbitMessage.getMessageProperties().getDeliveryTag(), false);
                    return;
                }
            }
            // 否则投递正常队列
            String routingKey = Constants.topic_push_7d_msg_que_transfer + id % 10;
            topicMqRabbitTemplate.convertAndSend(routingKey,JSONUtil.toJsonStr(eventPlanData));
            log.info("【转发处理】id：{}投递正常队列{}",id,routingKey);
            channel.basicAck(rabbitMessage.getMessageProperties().getDeliveryTag(), false);
        }catch (JSONException jsonException){
            log.error("json解析异常，消息: {}" ,JSONUtil.toJsonStr(message));
        }catch (Exception e) {
            e.printStackTrace();
            try {
                int retryCount = (int) rabbitMessage.getMessageProperties().getHeaders().getOrDefault("x-death", 0);
                if (retryCount > 5) {
                    // 重试次数超过阈值，选择丢弃
                    channel.basicNack(rabbitMessage.getMessageProperties().getDeliveryTag(), false, false);
                    log.error("消息推送数据重试次数超过阈值，丢弃消息: {}" + rabbitMessage.getMessageProperties().getMessageId());
                } else {
                    // 否则继续重试
                    channel.basicNack(rabbitMessage.getMessageProperties().getDeliveryTag(), false, true);
                }
            } catch (IOException ex) {
                ex.printStackTrace();
            }
        }
    }

    @RabbitListener(queues = Constants.topic_push_7d_msg_que8,containerFactory = "sevenDayMqFactory")
    public void keywordReceiveMessage8(String message, Channel channel, Message rabbitMessage) {

        try {
            EventPlanData eventPlanData = JSON.parseObject(message, EventPlanData.class);
            // 根据push7dId查询话题id
            LambdaQueryWrapper<TopicOrgan> topicOrganLambdaQueryWrapper = Wrappers.<TopicOrgan>lambdaQuery().eq(TopicOrgan::getPush7dId, eventPlanData.getTopicID());
            TopicOrgan topicOrgan = topicOrganService.getOne(topicOrganLambdaQueryWrapper);
            if(Objects.isNull(topicOrgan)){
                channel.basicAck(rabbitMessage.getMessageProperties().getDeliveryTag(), false);
                    return;
            }
            eventPlanData.setIs7d(1);
            Integer id = topicOrgan.getId();
            String queueName = Constants.TOPIC_PUSH_QUEUE + id;
            // 查询缓存中的队列名是否存在 并且可投递 是则投递临时队列
            String obj = String.valueOf(redisTemplate.opsForHash().get(Constants.TOPIC_PUSH_QUEUE_TEMP, queueName));
            TopicPushQueueTempCacheDTO topicPushQueueTempCacheDTO = null;
            if (StrUtil.isNotBlank(obj)) {
                ObjectMapper objectMapper = new ObjectMapper();
                try {
                    topicPushQueueTempCacheDTO = objectMapper.readValue(obj, TopicPushQueueTempCacheDTO.class);
                } catch (Exception e) {
                    // 处理异常
                    e.printStackTrace();
                    log.error("【临时队列数据序列化失败】：{}",e.getMessage());
                }
                if (Objects.nonNull(topicPushQueueTempCacheDTO) && topicPushQueueTempCacheDTO.getIsDelivery()) {
                    String queueName1 = topicPushQueueTempCacheDTO.getQueueName();
                    topicMqRabbitTemplate.convertAndSend(queueName1,JSONUtil.toJsonStr(eventPlanData));
                    log.info("【转发处理】id：{}投递临时队列{}",id,queueName1);

                    channel.basicAck(rabbitMessage.getMessageProperties().getDeliveryTag(), false);
                    return;
                }
            }
            // 否则投递正常队列
            String routingKey = Constants.topic_push_7d_msg_que_transfer + id % 10;
            topicMqRabbitTemplate.convertAndSend(routingKey,JSONUtil.toJsonStr(eventPlanData));
            log.info("【转发处理】id：{}投递正常队列{}",id,routingKey);
            channel.basicAck(rabbitMessage.getMessageProperties().getDeliveryTag(), false);
        }catch (JSONException jsonException){
            log.error("json解析异常，消息: {}" ,JSONUtil.toJsonStr(message));
        }catch (Exception e) {
            e.printStackTrace();
            try {
                int retryCount = (int) rabbitMessage.getMessageProperties().getHeaders().getOrDefault("x-death", 0);
                if (retryCount > 5) {
                    // 重试次数超过阈值，选择丢弃
                    channel.basicNack(rabbitMessage.getMessageProperties().getDeliveryTag(), false, false);
                    log.error("消息推送数据重试次数超过阈值，丢弃消息: {}" + rabbitMessage.getMessageProperties().getMessageId());
                } else {
                    // 否则继续重试
                    channel.basicNack(rabbitMessage.getMessageProperties().getDeliveryTag(), false, true);
                }
            } catch (IOException ex) {
                ex.printStackTrace();
            }
        }
    }

    @RabbitListener(queues = Constants.topic_push_7d_msg_que9,containerFactory = "sevenDayMqFactory")
    public void keywordReceiveMessage9(String message, Channel channel, Message rabbitMessage) {

        try {
            EventPlanData eventPlanData = JSON.parseObject(message, EventPlanData.class);
            // 根据push7dId查询话题id
            LambdaQueryWrapper<TopicOrgan> topicOrganLambdaQueryWrapper = Wrappers.<TopicOrgan>lambdaQuery().eq(TopicOrgan::getPush7dId, eventPlanData.getTopicID());
            TopicOrgan topicOrgan = topicOrganService.getOne(topicOrganLambdaQueryWrapper);
            if(Objects.isNull(topicOrgan)){
                channel.basicAck(rabbitMessage.getMessageProperties().getDeliveryTag(), false);
                    return;
            }
            eventPlanData.setIs7d(1);
            Integer id = topicOrgan.getId();
            String queueName = Constants.TOPIC_PUSH_QUEUE + id;
            // 查询缓存中的队列名是否存在 并且可投递 是则投递临时队列
            String obj = String.valueOf(redisTemplate.opsForHash().get(Constants.TOPIC_PUSH_QUEUE_TEMP, queueName));
            TopicPushQueueTempCacheDTO topicPushQueueTempCacheDTO = null;
            if (StrUtil.isNotBlank(obj)) {
                ObjectMapper objectMapper = new ObjectMapper();
                try {
                    topicPushQueueTempCacheDTO = objectMapper.readValue(obj, TopicPushQueueTempCacheDTO.class);
                } catch (Exception e) {
                    // 处理异常
                    e.printStackTrace();
                    log.error("【临时队列数据序列化失败】：{}",e.getMessage());
                }
                if (Objects.nonNull(topicPushQueueTempCacheDTO) && topicPushQueueTempCacheDTO.getIsDelivery()) {
                    String queueName1 = topicPushQueueTempCacheDTO.getQueueName();
                    topicMqRabbitTemplate.convertAndSend(queueName1,JSONUtil.toJsonStr(eventPlanData));
                    log.info("【转发处理】id：{}投递临时队列{}",id,queueName1);

                    channel.basicAck(rabbitMessage.getMessageProperties().getDeliveryTag(), false);
                    return;
                }
            }
            // 否则投递正常队列
            String routingKey = Constants.topic_push_7d_msg_que_transfer + id % 10;
            topicMqRabbitTemplate.convertAndSend(routingKey,JSONUtil.toJsonStr(eventPlanData));
            log.info("【转发处理】id：{}投递正常队列{}",id,routingKey);
            channel.basicAck(rabbitMessage.getMessageProperties().getDeliveryTag(), false);
        }catch (JSONException jsonException){
            log.error("json解析异常，消息: {}" ,JSONUtil.toJsonStr(message));
        }catch (Exception e) {
            e.printStackTrace();
            try {
                int retryCount = (int) rabbitMessage.getMessageProperties().getHeaders().getOrDefault("x-death", 0);
                if (retryCount > 5) {
                    // 重试次数超过阈值，选择丢弃
                    channel.basicNack(rabbitMessage.getMessageProperties().getDeliveryTag(), false, false);
                    log.error("消息推送数据重试次数超过阈值，丢弃消息: {}" + rabbitMessage.getMessageProperties().getMessageId());
                } else {
                    // 否则继续重试
                    channel.basicNack(rabbitMessage.getMessageProperties().getDeliveryTag(), false, true);
                }
            } catch (IOException ex) {
                ex.printStackTrace();
            }
        }
    }



}
